package cn.mesmile.flink.source;

import cn.mesmile.flink.jdkstream.VideoOrder;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author zb
 * @date 2021/8/21 16:56
 * @Description
 */
public class FlinkCustomSourceWebUIDemo04 {

    public static void main(String[] args) throws Exception {
        // 构建执行任务环境以及任务的启动的入口, 存储全局相关的参数
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 创建本地 UI 界面操作  127.0.0.1:8081
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(2);

        // 本机默认并行数为 12  ---> 本机配置为 6 核 12 线程
        VideoOrderSource videoOrderSource = new VideoOrderSource();

        DataStream<VideoOrder> videoOrderDataStream = env.addSource(videoOrderSource);
        videoOrderDataStream.filter(new FilterFunction<VideoOrder>() {
            @Override
            public boolean filter(VideoOrder value) throws Exception {
                return value.getMoney() > 5;
            }
        }).setParallelism(3);


        videoOrderDataStream.print().setParallelism(4);

        //DataStream需要调用execute,可以取个名称
        env.execute("custom source job");
    }
}
